
#include "Worker.h"

void  Async_Producer(CMsgTyped* pmsg,  sql::Statement* pConn);
void  Async_Consumer(CMsgTyped* pmsg,  sql::Statement* pConn);

char data[ 1024 * 1024];
char g_sql[1024*1024*4];
int op_cnt = 0;

void PrintCnt(bool bforce)
{
    static time_t last = 0;
    op_cnt ++;

    time_t n = time( NULL );

    if (n - last > 1 || bforce)
    {
        printf("已执行sql数量%d。\n", op_cnt );
        last = n;
    }
}

void ReadBlobString(string& out, sql::ResultSet* row, const char* name)
{
    istream* pstream = row->getBlob("save_data");

    pstream->seekg(0, ios_base::end);
    int size = pstream->tellg();
    pstream->seekg(0, ios_base::beg);

    if (size > 0 && size < 1024*1024)
    {
        pstream->read( data, size);
        BinaryToText( out, data, size);
    }

    delete pstream;
}

string DataToString_GlobalRoles(sql::ResultSet* row)
{
    /*roleId INT(16) UNSIGNED NOT NULL AUTO_INCREMENT,   d
	name VARCHAR(64) NOT NULL,                           s
	account VARCHAR(64) NOT NULL,                        s
	role_type INT(10) NOT NULL,                          d
	sex INT(10) NULL DEFAULT NULL,                       d
	create_time DATETIME NULL DEFAULT NULL,              s
	save_loc INT(10) NULL DEFAULT NULL,                  d
	logintime DATETIME NULL DEFAULT NULL,                s
	usr_crt_dt DATETIME NULL DEFAULT NULL,*/                          //s

	sprintf(g_sql, "insert into GlobalRoles values('%d', '%s', '%s', '%d', '%d','%s','%d','%s', '%s')",
        row->getInt(1),
         row->getString(2).c_str(),
         row->getString(3).c_str(),
         row->getInt(4),
         row->getInt(5),
         row->getString(6).c_str(),
         row->getInt(7),

         row->getString(8).c_str(),
         row->getString(9).c_str()
         );

    return g_sql;
}

string DataToString_GameRoles(sql::ResultSet* row)
{
    /*name VARCHAR(64) NOT NULL,                         s
	account VARCHAR(64) NOT NULL,                        s
	role_type INT(10) NOT NULL,                          d
	create_time DATETIME NULL DEFAULT NULL,              s
	server_id INT(10) NULL DEFAULT NULL,                 d
	total_sec INT(10) NULL DEFAULT NULL,                 d
	role_index INT(10) NULL DEFAULT NULL,                d
	scene_id INT(10) NULL DEFAULT NULL,                  d
	posi_x FLOAT NULL DEFAULT NULL,                      f
	posi_y FLOAT NULL DEFAULT NULL,                      f
	save_para VARCHAR(2000) NULL DEFAULT NULL,           s
	save_time DATETIME NULL DEFAULT NULL,                s
	save_data MEDIUMBLOB NULL,                           s
	roleId INT(10) NOT NULL,                             d
	*/

    string s;
    ReadBlobString(s, row, "save_data");
	string res = "0x";
	if (s.size()) res += s;
	else res = "NULL";

	sprintf(g_sql, "insert into GameRoles values('%s', '%s', '%d', '%s', '%d','%d','%d','%d', '%f', '%f', '%s', '%s', %s, '%d')",
        row->getString(1).c_str(),
         row->getString(2).c_str(),
         row->getInt(3),
         row->getString(4).c_str(),
         row->getInt(5),
         row->getInt(6),
         row->getInt(7),
         row->getInt(8),

        (float)row->getDouble(9),
        (float) row->getDouble(10),

         row->getString(11).c_str(),
         row->getString(12).c_str(),
         res.c_str() ,
         row->getInt(14)
         );

    //Log(LOG_NORMAL, "%s\n", g_sql);

    return g_sql;
}

string DataToString_FamilyInfo(sql::ResultSet* row)
{
    	/*FamilyName VARCHAR(64) NOT NULL COMMENT '工会名称',                       s
	FamilyExp FLOAT NOT NULL COMMENT '工会活力',                                                f
	FamilyLevel INT(11) NOT NULL COMMENT '等级',                                               d
	FamilyBulletin VARCHAR(105) NULL DEFAULT NULL COMMENT '公告',        s
	LastVisitTime DATETIME NULL DEFAULT NULL COMMENT '最后活跃时间',    s
	CreateTime DATETIME NOT NULL COMMENT '创建时间',                                        s
	Guid VARCHAR(64) NULL DEFAULT NULL COMMENT '唯一标识符',*/              //s

		sprintf(g_sql, "insert into FamilyInfo values('%s', '%f', '%d', '%s', '%s','%s','%s')",
        row->getString(1).c_str(),
         float( row->getDouble(2) ),
         row->getInt(3),

         row->getString(4).c_str(),
         row->getString(5).c_str(),
         row->getString(6).c_str(),
         row->getString(7).c_str()
         );

    return g_sql;
}

string DataToString_FamilyMember(sql::ResultSet* row)
{
    /*
  MemberName VARCHAR(64) NOT NULL,
	FamilyName VARCHAR(64) NOT NULL,
	MemberJob INT(11) NOT NULL COMMENT '职位权限',
	MemberContribute FLOAT NOT NULL COMMENT '工会总贡献',
	JoinTime DATETIME NOT NULL COMMENT '入会时间',
	JobDescribe VARCHAR(64) NULL DEFAULT NULL COMMENT '职位名称',
	PigCon INT(11) NULL DEFAULT '0' COMMENT '喂猪的贡献',
	*/

        sprintf(g_sql, "insert into FamilyMember values('%s', '%s', '%d', '%f', '%s','%s','%d')",
         row->getString(1).c_str(),
         row->getString(2).c_str(),
         row->getInt(3),

         float( row->getDouble(4) ),
         row->getString(5).c_str(),
         row->getString(6).c_str(),
         row->getInt(7)
         );

    return g_sql;
}

string DataToString_MasterInfo(sql::ResultSet* row)
{
    /*
    MasterName VARCHAR(64) NOT NULL,
	InsertTime DATETIME NULL DEFAULT NULL,*/

        sprintf(g_sql, "insert into MasterInfo values('%s', '%s')",
         row->getString(1).c_str(),
         row->getString(2).c_str()
         );

    return g_sql;
}

string DataToString_StudentInfo(sql::ResultSet* row)
{
    /*
  StudentName VARCHAR(64) NOT NULL,
	MasterName VARCHAR(64) NOT NULL,
	JoinTime DATETIME NULL DEFAULT NULL,*/

    sprintf(g_sql, "insert into StudentInfo values('%s', '%s', '%s')",
         row->getString(1).c_str(),
         row->getString(2).c_str(),
         row->getString(3).c_str()
         );

    return g_sql;
}
string DataToString_MailTable(sql::ResultSet* row)
{
    /*
    MailID CHAR(64) NOT NULL,
	Owner VARCHAR(64)  NULL,
	Title VARCHAR(64)  NULL
	Sender VARCHAR(64)  NULL,
	Receiver VARCHAR(64)  NULL
	IsReaded INT NULL
	SendTime DATETIME,
	Attachment VARCHAR(64) NULL,
	MailContent NOT NULL,*/

    sprintf(g_sql, "insert into MailTable values('%s', '%s', '%s', '%s', '%s', '%d', '%s', '%s', '%s')",
         row->getString(1).c_str(),
         row->getString(2).c_str(),
         row->getString(3).c_str(),
         row->getString(4).c_str(),
         row->getString(5).c_str(),
         row->getInt(6),
         row->getString(7).c_str(),
         row->getString(8).c_str(),
         row->getString(9).c_str()
         );

    return g_sql;
}
string DataToString(const char* tableName , sql::ResultSet* row)
{
    if (strcmp(tableName, "GlobalRoles") == 0) return DataToString_GlobalRoles(row);
    if (strcmp(tableName, "GameRoles") == 0) return DataToString_GameRoles(row);
    if (strcmp(tableName, "FamilyInfo") == 0) return DataToString_FamilyInfo(row);
    if (strcmp(tableName, "FamilyMember") == 0) return DataToString_FamilyMember(row);
    if (strcmp(tableName, "MasterInfo") == 0) return DataToString_MasterInfo(row);
    if (strcmp(tableName, "StudentInfo") == 0) return DataToString_StudentInfo(row);
    if (strcmp(tableName, "MailTable") == 0)
        return DataToString_MailTable(row);

    return "";
}

// 任务列表
class SqlList : public CriticalObject
{
public:
    void push( string& str )
    {
        SMART_LOCK_THIS();
        m_sqls.push_back( str );
    }

    void push( const char* str )
    {
        SMART_LOCK_THIS();
        if (str) m_sqls.push_back( str );
    }

    string pop()
    {
        SMART_LOCK_THIS();

        if (m_sqls.size() == 0) return "";
        string ret = m_sqls[0];
        m_sqls.erase( m_sqls.begin() );
        return ret;
    }

    size_t size()
    {
        SMART_LOCK_THIS();
        return m_sqls.size();
    }

    void clear()
    {
        SMART_LOCK_THIS();
        m_sqls.clear();
    }

    vector<string> m_sqls;
};

SqlList g_sqls;

// 计数器
Counter g_jionCounter;

void Worker::DoJionTable(const char* tableName, int from_db, int to_db)
{
    if (!tableName) return;

    printf("\n合并%s\n",  tableName);
    printf("------------------------------------\n");

    // 创建生产者和消费者
    g_sqls.clear();
    g_jionCounter.Set( 1 );

    CMsgTyped param;
    param.SetString( tableName);
    _PROC->Call(from_db, Async_Producer, &param);
    _PROC->Call(to_db, Async_Consumer,  NULL);

    g_jionCounter.Wait0();
}

void  Async_Producer(CMsgTyped* pmsg,  sql::Statement* pConn)
{
    if (!pmsg) return;

    const char* db = pmsg->StringVal();

    char sql[256];
    sprintf(sql, "select * from %s", db);

    MysqlResult res;
	try
	{
		res.dataSet = pConn->executeQuery(sql);

		//  printf(">>>> %s <<< \n", sql );
		if (res.dataSet)
		{
            while (res.dataSet->next())
            {
                if (g_sqls.size() > 1000)
                {
                    // 等1秒
                    usleep( 100000 );
                }

                // 插入任务
                string sql = DataToString(db,  res.dataSet);
                g_sqls.push( sql );
            }
		}
	}
	catch(sql::SQLException & e)
    {
        printf(">>>> %s <<< \n", sql );

        cout << __LINE__ << "\n";
        cout << "# ERR: " << e.what() << "\n";
    }

    g_sqls.push("exit");
}

void  Async_Consumer(CMsgTyped* pmsg,  sql::Statement* pConn)
{
    while(1)
    {
        int cnt = g_sqls.size();
        if (cnt == 0 )
        {
            // 等待0.01秒
            usleep( 10000 );
            continue;
        }

        // 执行sql
        string sql = g_sqls.pop( );
        if (sql.size() == 0) continue;

       // printf(">>>> %s <<< \n", sql.c_str() );

        if (sql == "exit") break;

        try
        {
            pConn->executeUpdate(sql.c_str());
            PrintCnt( false );
        }
        catch(sql::SQLException & e)
        {
            //cout << __LINE__ << "\n";
          //  cout << "# ERR: " << e.what() << "\n";

           //  printf(">>>> %s <<< \n", sql.c_str() );
        }
    }

    // 退出整个工作
    g_jionCounter.Set( 0 );
}



